SparkSQL的自适应执行-Adaptive Execution
点击上方蓝色字体,选择“设为星标”
Spark SQL 自适应执行优化引擎
背景
Adaptive Execution 将可以根据执行过程中的中间数据优化后续执行,从而提高整体执行效率。核心在于两点
执行计划可动态调整
调整的依据是中间结果的精确统计信息
spark 2.3 开始试验功能
spark 3.0 正式发布 自适应查询执行(Adaptive Query Execution)
如何设置合适的shuffle partition数量?
在Spark SQL中, shufflepartition数可以通过参数spark.sql.shuffle.partition来设置,默认值是200。
如果partition太小,单个任务处理的数据量会越大,在内存有限的情况,就会写文件,降低性能,还会oom
如果partition太大,每个处理任务数据量很小,很快结束,导致spark调度负担变大,中间临时文件多
spark sql 最佳执行计划
Spark SQL的Catalyst优化器的核心工作就是选择最佳的执行计划,主要依靠:
早起基于规则的优化器RBO
spark2.2 加入基于代价的优化CBO
执行计划在计划阶段确定后,不会改变,如果能够获取运行时信息,就可能得到一个更加的执行计划
数据倾斜如何处理
数据倾斜是指某一个partition的数据量远远大于其它partition的数据,导致个别任务的运行时间远远大于其它任务,因此拖累了整个SQL的运行时间。
常见手段:
增加shuffle partition数量,让热点partition的数据分散一些,但是对于同一个key没有作用
增加 BroadcastHashJoin的阈值,在某些场景下可以把SortMergeJoin转化成BroadcastHashJoin而避免shuffle产生的数据倾斜。
手动过滤倾斜key,加入前缀,join表也对key膨胀处理,再join
spark 能否运行时自动处理join中的数据倾斜
自适应执行架构
sql -> 解析 -> 逻辑计划 -> 物理计划 -> rdd -> job -> dag -> stage -> task run
一旦执行计划确定,无法更新
按照每个reducer处理partition数据内存大小分,每个64m
按照每个reducer处理partition数据条数分,100000条
动态调整执行计划
在运行时动态调整join的策略,在满足条件的情况下,即一张表小于Broadcast阈值,可以将SortMergeJoin转化成BroadcastHashJoin。
SortMergeJoin,每个reducer通过网络shuffle读取属于自己的数据;会出现不同程度的数据倾斜问题;
BroadcastHashJoin,每一个reducer读取一个mapper的整个shuffle output文件,shuffle读变成了本地读取,没有数据通过网络传输;数据量一般比较均匀,也就避免了倾斜;
动态处理数据倾斜
在运行时很容易地检测出有数据倾斜的partition,当执行某个stage时,我们收集该stage每个mapper 的shuffle数据大小和记录条数
如果某一个partition的数据量或者记录条数超过中位数的N倍,并且大于某个预先配置的阈值,我们就认为这是一个数据倾斜的partition,需要进行特殊的处理
Spark 使用
配置参数
org.apache.spark.sql.internal.SQLConf
spark.sql.adaptive.enabled=true
倾斜处理开关
spark.sql.adaptive.shuffle.targetPostShuffleInputSize
动态调整 reduce 个数的 partition 大小依据。如设置 64MB,则 reduce 阶段每个 task 最少处理 64MB 的数据。默认值为 64MB。
spark.sql.adaptive.minNumPostShufflePartitions -- v2.4 有 3.0 已经去掉
动态调整 reduce 个数的 partition 条数依据。如设置 20000000,则 reduce 阶段每个 task 最少处理 20000000 条的数据。默认值为 20000000。
spark.sql.adaptive.forceApply -- V3.0
自适应执行在没有需要shuffle或者子查询的时候将不适用,当设为true始终使用
spark.sql.adaptive.logLevel --v3.0
自适应执行时产生的日志等级
spark.sql.adaptive.advisoryPartitionSizeInBytes -- v3.0
倾斜数据分区拆分,小数据分区合并优化时,建议的分区大小
与spark.sql.adaptive.shuffle.targetPostShuffleInputSize含义相同
spark.sql.adaptive.coalescePartitions.enabled -- v3.0
是否开启合并小数据分区默认开启,调优策略之一
spark.sql.adaptive.coalescePartitions.minPartitionNum -- v3.0
合并后最小的分区数
spark.sql.adaptive.fetchShuffleBlocksInBatch -- v3.0
是否批量拉取blocks,而不是一个个的去取
给同一个map任务一次性批量拉取blocks可以减少io 提高性能
spark.sql.adaptive.skewJoin.enabled
自动倾斜处理,处理 sort-merge join中的倾斜数据
spark.sql.adaptive.skewJoin.skewedPartitionFactor
判断分区是否是倾斜分区的比例
当一个 partition 的 size 大小大于该值(所有 parititon 大小的中位数)且大于spark.sql.adaptive.skewedPartitionSizeThreshold,或者 parition 的条数大于该值(所有 parititon 条数的中位数)且大于 spark.sql.adaptive.skewedPartitionRowCountThreshold,才会被当做倾斜的 partition 进行相应的处理。默认值为 10
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes
单个分区大于默认256MB
参考
https://issues.apache.org/jira/browse/SPARK-23128
https://blog.csdn.net/weixin_34006468/article/details/91894261
https://www.cnblogs.com/zz-ksw/p/11254294.html
文章不错?点个【在看】吧! 👇